package drds.binlog.instance.core;

import drds.binlog.common.AbstractLifeCycle;
import drds.binlog.common.Authors;
import drds.binlog.common.alarm.AlarmHandler;
import drds.binlog.common.position.ClientInfo;
import drds.binlog.filter.aviater.AviaterRegexFilter;
import drds.binlog.metadata.MetaDataManager;
import drds.binlog.parse.EventParser;
import drds.binlog.parse.ha.HAController;
import drds.binlog.parse.ha.HeartBeatHAController;
import drds.binlog.parse.inbound.AbstractEventParser;
import drds.binlog.parse.inbound.group.GroupEventParser;
import drds.binlog.parse.inbound.mysql.MysqlEventParser;
import drds.binlog.parse.index.LogPositionManager;
import drds.binlog.protocol.entry_protocol_proto.Entry;
import drds.binlog.sink.EventSink;
import drds.binlog.store.EventStore;
import drds.binlog.store.model.Event;
import drds.common.Author;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

@Author(name = Authors.LI_YANG)
public class AbstractBinlogTask extends AbstractLifeCycle implements BinlogTask
{

    private static final Logger logger = LoggerFactory.getLogger(AbstractBinlogTask.class);

    protected Long binlogTaskId;                                                      // 和manager交互唯一标示
    protected String destination;                                                  // 队列名字
    protected MetaDataManager metaDataManager;                                                  // 消费信息管理器
    protected AlarmHandler alarmHandler;
    //
    protected EventParser eventParser;                                                  // 解析对应的数据信息
    protected EventSink<List<Entry>> eventSink;                                                    // 链接parse和store的桥接器

    protected EventStore<Event> eventStore;                                                   // 有序队列
    // alarm报警机制

    @Override
    public boolean subscribeChange(ClientInfo clientInfo)
    {
        if (StringUtils.isNotEmpty(clientInfo.getFilter()))
        {
            logger.info("subscribe eventFilter change to " + clientInfo.getFilter());
            AviaterRegexFilter aviaterRegexFilter = new AviaterRegexFilter(clientInfo.getFilter());

            boolean isGroup = (eventParser instanceof GroupEventParser);
            if (isGroup)
            {
                // 处理group的模式
                List<EventParser> eventParserList = ((GroupEventParser) eventParser).getEventParserList();
                for (EventParser eventParser : eventParserList)
                {// 需要遍历启动
                    ((AbstractEventParser) eventParser).setEventFilter(aviaterRegexFilter);
                }
            } else
            {
                ((AbstractEventParser) eventParser).setEventFilter(aviaterRegexFilter);
            }

        }

        // filter的处理规则
        // a. parser处理数据过滤处理
        // b. sink处理数据的路由&分发,一份parse数据经过sink后可以分发为多份，每份的数据可以根据自己的过滤规则不同而有不同的数据
        // 后续内存版的一对多分发，可以考虑
        return true;
    }

    @Override
    public void start()
    {
        super.start();
        if (!metaDataManager.isRunning())
        {
            metaDataManager.start();
        }

        if (!alarmHandler.isRunning())
        {
            alarmHandler.start();
        }
        //
        if (!eventStore.isRunning())
        {
            eventStore.start();
        }

        if (!eventSink.isRunning())
        {
            eventSink.start();
        }

        if (!eventParser.isRunning())
        {
            beforeStartEventParser(eventParser);
            eventParser.start();
            afterStartEventParser(eventParser);
        }
        logger.info("start successful....");
    }

    @Override
    public void stop()
    {
        super.stop();
        logger.info("stop CannalInstance for {}-{} ", new Object[]{binlogTaskId, destination});

        if (eventParser.isRunning())
        {
            beforeStopEventParser(eventParser);
            eventParser.stop();
            afterStopEventParser(eventParser);
        }

        if (eventSink.isRunning())
        {
            eventSink.stop();
        }

        if (eventStore.isRunning())
        {
            eventStore.stop();
        }
        //
        if (metaDataManager.isRunning())
        {
            metaDataManager.stop();
        }

        if (alarmHandler.isRunning())
        {
            alarmHandler.stop();
        }

        logger.info("stop successful....");
    }

    protected void beforeStartEventParser(EventParser eventParser)
    {

        boolean isGroup = (eventParser instanceof GroupEventParser);
        if (isGroup)
        {
            // 处理group的模式
            List<EventParser> eventParserList = ((GroupEventParser) eventParser).getEventParserList();
            for (EventParser $eventParser : eventParserList)
            {// 需要遍历启动
                startEventParserInternal($eventParser, true);
            }
        } else
        {
            startEventParserInternal(eventParser, false);
        }
    }

    // around event parser, default impl
    protected void afterStartEventParser(EventParser eventParser)
    {
        // 读取一下历史订阅的filter信息
        List<ClientInfo> clientInfoList = metaDataManager.getClientInfoList(destination);
        for (ClientInfo clientInfo : clientInfoList)
        {
            subscribeChange(clientInfo);
        }
    }

    // around event parser
    protected void beforeStopEventParser(EventParser eventParser)
    {
        // noop
    }

    protected void afterStopEventParser(EventParser eventParser)
    {

        boolean isGroup = (eventParser instanceof GroupEventParser);
        if (isGroup)
        {
            // 处理group的模式
            List<EventParser> eventParserList = ((GroupEventParser) eventParser).getEventParserList();
            for (EventParser $eventParser : eventParserList)
            {// 需要遍历启动
                stopEventParserInternal($eventParser);
            }
        } else
        {
            stopEventParserInternal(eventParser);
        }
    }

    /**
     * 初始化单个eventParser，不需要考虑group
     */
    protected void startEventParserInternal(EventParser eventParser, boolean isGroup)
    {
        if (eventParser instanceof AbstractEventParser)
        {
            AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
            // 首先启动log position管理器
            LogPositionManager logPositionManager = abstractEventParser.getLogPositionManager();
            if (!logPositionManager.isRunning())
            {
                logPositionManager.start();
            }
        }

        if (eventParser instanceof MysqlEventParser)
        {
            MysqlEventParser mysqlEventParser = (MysqlEventParser) eventParser;
            HAController haController = mysqlEventParser.getHaController();

            if (haController instanceof HeartBeatHAController)
            {
                ((HeartBeatHAController) haController).setCanalHASwitchable(mysqlEventParser);
            }

            if (!haController.isRunning())
            {
                haController.start();
            }

        }
    }

    protected void stopEventParserInternal(EventParser eventParser)
    {
        if (eventParser instanceof AbstractEventParser)
        {
            AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
            // 首先启动log position管理器
            LogPositionManager logPositionManager = abstractEventParser.getLogPositionManager();
            if (logPositionManager.isRunning())
            {
                logPositionManager.stop();
            }
        }

        if (eventParser instanceof MysqlEventParser)
        {
            MysqlEventParser mysqlEventParser = (MysqlEventParser) eventParser;
            HAController haController = mysqlEventParser.getHaController();
            if (haController.isRunning())
            {
                haController.stop();
            }
        }
    }


    @Override
    public String getDestination()
    {
        return destination;
    }

    @Override
    public MetaDataManager getMetaDataManager()
    {
        return metaDataManager;
    }

    @Override
    public AlarmHandler getAlarmHandler()
    {
        return alarmHandler;
    }

    @Override
    public EventParser getEventParser()
    {
        return eventParser;
    }

    @Override
    public EventSink getEventSink()
    {
        return eventSink;
    }

    @Override
    public EventStore getEventStore()
    {
        return eventStore;
    }


}
